package com.edu.realtime.app.func;

import java.util.Collections;
import java.util.concurrent.ExecutorService;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import com.alibaba.fastjson.JSONObject;
import com.edu.realtime.util.DimUtil;
import com.edu.realtime.util.ThreadPoolUtil;

/**
 * Created on 2022/10/20.
 *
 * @author Topus
 */
public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T, T> {
    // 实现异步请求指定的表名
    private String tableName;

    public DimAsyncFunction(String tableName) {
        this.tableName = tableName;
    }

    // 定义线程池对象
    private ExecutorService executorService;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 获取线程池
        executorService = ThreadPoolUtil.getInstance();
    }

    @Override
    public void asyncInvoke(T input, ResultFuture<T> resultFuture) {
        // 模板方法嵌套
        executorService.submit(new Runnable() {
            @Override
            public void run() {
                // 1）获取主键
                String key = getKey(input);
                // 2）根据主键获取维度对象
                JSONObject dimInfo = DimUtil.getDimInfo(tableName, key);
                // 3）维度对象属性填充
                if (dimInfo != null) {
                    join(input, dimInfo);
                }
                // 4）获取回调结果传送到下游
                resultFuture.complete(Collections.singleton(input));
            }
        });
    }

    // 关联维度数据，填充input
    public abstract void join(T input, JSONObject dimInfo);

    // 根据传入参数获取key=>id
    public abstract String getKey(T obj);
}
